package org.apache.rocketmq.example.kang.consumer.multiDataCenter;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMachineRoomNearby;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingUtil;

import java.util.List;

public class MultiDataCenterConsumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new
                DefaultMQPushConsumer("dw_test_consumer_6");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setClientIP("MachineRoom1-" + RemotingUtil.getLocalAddress());
        // consumer.setClientIP("MachineRoom2-" + RemotingUtil.getLocalAddress());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe("machine_topic_test", "*");
        AllocateMessageQueueAveragely averagely = new AllocateMessageQueueAveragely();
        AllocateMachineRoomNearby.MachineRoomResolver machineRoomResolver =  new
                AllocateMachineRoomNearby.MachineRoomResolver() {
                    @Override
                    public String brokerDeployIn(MessageQueue messageQueue) {
                        return messageQueue.getBrokerName().split("-")[0];
                    }
                    @Override
                    public String consumerDeployIn(String clientID) {
                        return clientID.split("-")[0];
                    }
                };

        consumer.setAllocateMessageQueueStrategy(new
                AllocateMachineRoomNearby(averagely, machineRoomResolver));
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                try {
                    System.out.printf("%s Receive New Messages: %s %n",
                            Thread.currentThread().getName(), msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Throwable e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }

}
